package com.lagou.bak;

import org.apache.commons.math3.fitting.leastsquares.EvaluationRmsChecker;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.TableSchema;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.catalog.Catalog;
import org.apache.flink.table.catalog.ObjectPath;
import org.apache.flink.types.Row;

import java.util.Arrays;
import java.util.List;
import java.util.Optional;
import java.util.Random;

import static org.apache.flink.table.api.Expressions.$;

public class Six {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        DataStreamSource<Area> areaDataStreamSource = env.addSource(new SourceFunction<Area>() {
            @Override
            public void run(SourceContext<Area> ctx) throws Exception {
                // String[] words = {"hello", "spark", "fink", "hadoop", "hbase", "impala", "hive", "spark", "mr"};
//                (110107,010-39.9056-3-116.223-中国,北京,北京市,石景山区-石景山区-110100-Shijingshan-石景山-100043)
//                (110116,010-40.316-3-116.632-中国,北京,北京市,怀柔区-怀柔区-110100-Huairou-怀柔-101400)
                while (true) {
                    Random random = new Random();
//                    int i = random.nextInt(words.length);
                    Area info = new Area(110107, "石景山", 1000100);
                    Area info1 = new Area(1000100, "北京", 10001009);
                    int i = random.nextInt(2);
                    if(i==1){
                        ctx.collect(info);
                    }else {
                        ctx.collect(info1);
                    }
                    Thread.sleep(1000);
                }
            }

            @Override
            public void cancel() {
            }
        });

        //创建临时视图
        tEnv.createTemporaryView("area",areaDataStreamSource,$("a"),$("b"),$("c"));
        tEnv.createTemporaryView("area1",areaDataStreamSource);
        tEnv.createTemporaryView("area2",areaDataStreamSource);
        //获取临时视图（有视图）
        String[] views = tEnv.listViews();

        System.out.println(Arrays.toString(views)); //[area, area1, area2]
        //获取数据库
        String currentDatabase = tEnv.getCurrentDatabase();
        System.out.println(currentDatabase); //default_database
        //获取catalog
        String currentCatalog = tEnv.getCurrentCatalog();
        System.out.println("currentCatalog:" + currentCatalog); //default_catalog



        Optional<Catalog> default_catalog = tEnv.getCatalog(currentCatalog);
        //通过catalog获取scheme（就是找不到area视图，明明前面都有）
        System.out.println("default_catalog:" + default_catalog); //default_catalog:Optional[org.apache.flink.table.catalog.GenericInMemoryCatalog@255990cc]

        Catalog catalog = default_catalog.get();
        boolean exists = catalog.databaseExists("default_database");
        System.out.println("default_database: " + exists);  //default_database: true

        boolean areaExists = catalog.tableExists(new ObjectPath("default_database","area"));
        System.out.println("areaExists: " + areaExists);  //areaExists: false

        List<String> listTables = catalog.listTables("default_database");
        System.out.println("listtables:" + listTables);  //listtables:[]

        List<String> listViews = catalog.listViews("default_database");
        System.out.println("listViews:" + listViews);  //listViews:[]

        String[] tables = tEnv.listTables();
        System.out.println("tables:" + Arrays.toString(tables));  //tables:[area, area1, area2]

        Table tArea = tEnv.sqlQuery("select * from area");
        TableSchema tAreaSchema = tArea.getSchema();
        System.out.println("tAreaSchema:" + tAreaSchema);

        String s = tAreaSchema.toRowDataType().toString();
        System.out.println("s:" + s);


        TableSchema area = default_catalog.get().getTable(new ObjectPath("default_database","area")).getSchema();
        System.out.println("areaSchema:" + area);
        String string = area.toRowDataType().toString();
        System.out.println(string);

        //Table select = tEnv.sqlQuery(sql);
        //DataStream<Tuple2<Boolean, Row>> retractStream = tEnv.toRetractStream(select, Row.class);
        //retractStream.print();
        //env.execute();
    }
}
